#include "util-internal.h"
#include "udpclient.h"
#include "asynevent.h"
#include "ntdtime.h"

#define TEMP_BUFFER		4096
#define SYN_INTERVAL    100000//100ms
#define SYN_TIMEOUT     3000000//3000ms

static void cb_func(evutil_socket_t fd, short what, void *user_data)
{
	if( what&EV_READ ){
		(static_cast<UDPClient *>(user_data))->OnRead();
	}

	if( what&EV_WRITE ){
		//(static_cast<UDPClient *>(user_data))->OnWrite();
		(static_cast<UDPClient *>(user_data))->OnTimer(NULL);
	}
}

//static void timeout_cb(evutil_socket_t fd, short event, void *user_data)
//{
//	//(static_cast<UDPClient *>(user_data))->OnTimer();
//	(static_cast<UDPClient *>(user_data))->OnWrite();
//}

UDPClient::UDPClient(IUdpClientListener* pListen)
{
	m_base = AsynEventFramework::Instance()->GetEventBase();

	//m_seq = RANDOM_32();

	//m_owd_base = 0;

	////Pkt size will be adjust to ratelimit, so move it to SetRateLimit.
	////m_protodata.m_len = MSS_SIZE;
	////m_protodata.m_pData = new byte[MSS_SIZE];
	////memset(m_protodata.m_pData, 0, m_protodata.m_len*sizeof(byte));

	//m_lastsent = 0;
	//m_hsInterval = SYN_INTERVAL;

	//SetRateLimit(UINT_MAX);

	/////* Initalize one event */
	////event_assign(&m_timeout, m_base, -1, EV_PERSIST, timeout_cb, this);

	////struct timeval tv;
	////evutil_timerclear(&tv);
	////tv.tv_sec = 1;
	////event_add(&m_timeout, &tv);

	//m_bInited = false;

	//m_protodata.m_convId = RANDOM_16();

	//m_total = 0;

	//m_bFin = false;

	m_total = 0;

	m_pListen = pListen;
	//event_assign(&m_timeout, m_base, -1, 0, timeout_cb, this);//no EV_PERSIST

	//event_assign(&m_timeout, m_base, -1, EV_PERSIST, timeout_cb, this);
	//timeval tv = {0, 10000};
	//event_add(&m_timeout, &tv);

	m_tick.AddListener(this);
	m_send_bucket.AddListener(this);


	Reset();

}

UDPClient::~UDPClient()
{
	m_tick.RemoveListener(this);
	m_send_bucket.RemoveListener(this);

	Close();

	delete []m_protodata.m_pData;
}

void UDPClient::Reset()
{
	m_seq = RANDOM_32();
	if( m_bPacketPair ){
		if(m_seq%2 != 0){
			++m_seq;//packet pair use even seq as the first packet
		}
	}

	m_owd_base = 0;

	m_lastsent = 0;
	m_hsInterval = SYN_INTERVAL;

	SetRateLimit(UINT_MAX);

	m_bInited = false;

	m_protodata.m_convId = RANDOM_16();

	m_bFin = false;

	m_per_total = 0;
	//m_totaltick = 0;

	m_bPacketPair = false;
}

bool UDPClient::Connect(const std::string& ip, int port)
{
	m_evbuf_input = evbuffer_new();
	m_evbuf_output = evbuffer_new();

	m_socket = socket( PF_INET, SOCK_DGRAM, IPPROTO_UDP );
	if ( m_socket == INVALID_SOCKET ) 
		return false;

	int RcvBufferSize = 1024*1024;
	setsockopt( m_socket, SOL_SOCKET, SO_RCVBUF, (char*)&RcvBufferSize, sizeof( RcvBufferSize ) );

	SetSockAddr(m_dest, ntohl(inet_addr(ip.c_str())), port);

	sockaddr_in saHost;
	saHost.sin_family = PF_INET;
	saHost.sin_addr.s_addr = INADDR_ANY;
	saHost.sin_port = 0;

	if( bind( m_socket, (sockaddr*)&saHost, sizeof(saHost)) == -1 ){
		return false;
	}

	evutil_make_socket_nonblocking(m_socket);

	//m_ev = event_new(m_base, m_socket, EV_WRITE|EV_READ|EV_PERSIST, cb_func,this);

	m_ev = event_new(m_base, m_socket, EV_READ|EV_PERSIST, cb_func,this);

	//timeval timeout = {0, 10000};
	event_add(m_ev, NULL);

	m_writeEv = event_new(m_base, m_socket, EV_WRITE, cb_func,this);
	event_add(m_writeEv, NULL);

	m_tick.Start(1, true);//1ms tick

	//event_assign(&m_timeout, m_base, -1, 0, timeout_cb, this);//no EV_PERSIST

	//SetDuration(g_settings.m_probeTime);
	//struct timeval tv;
	//evutil_timerclear(&tv);
	//tv.tv_sec = static_cast<long>(g_settings.m_probeTime/1);
	//tv.tv_usec = static_cast<long>((g_settings.m_probeTime - tv.tv_sec) * 1000000);
	////timeval tv = {0, 500000};
	//event_add(&m_timeout, &tv);

	//m_starttick = Now_us();//SHOULD NOT BE HERE, WHEN RTT IS LARGE, THE HANDSHAKE WILL COST SOME TIME!!

	if( m_pktInterval == 0 ){
		LOG4CPLUS_DEBUG(g_logger,"Sending back-to-back packet train.");
	}

	return true;
}

void UDPClient::SetDuration(float t)
{
	//evtimer_del(&m_timeout);

	//struct timeval tv;
	//evutil_timerclear(&tv);
	//tv.tv_sec = static_cast<long>(t/1);
	//tv.tv_usec = static_cast<long>((t - tv.tv_sec) * 1000000);
	////timeval tv = {0, 500000};
	//event_add(&m_timeout, &tv);

	//struct timeval tv;
	//evutil_timerclear(&tv);
	//uint64 tt = m_pktInterval/10;
	//tv.tv_sec = static_cast<long>(tt/1000000);
	//tv.tv_usec = static_cast<long>(tt%1000000);
	//tv.tv_sec = 0;
	//tv.tv_usec = 1000;
	//event_add(&m_timeout, &tv);

	//Should not be here, for there maybe deadlock caused by handshake lost and removed event_write
	//m_tick.Start(1, true);//1ms tick

	//LOG4CPLUS_DEBUG(g_logger, "Start 1ms tick");

	m_starttick = Now_us();

	m_timeout_us = 1000000*t;


	//LOG4CPLUS_DEBUG(g_logger, "add timeout: "<<t);
}

void UDPClient::Close()
{
	Reset();

	event_del(m_ev);
	event_free(m_ev);

	event_del(m_writeEv);
	event_free(m_writeEv);

	//event_del(&m_timeout);
	m_tick.Stop();

	evbuffer_free(m_evbuf_input);
	evbuffer_free(m_evbuf_output);

	m_evbuf_input = NULL;
	m_evbuf_output = NULL;
	m_ev = NULL;
	m_writeEv = NULL;
}

void UDPClient::OnFinish()
{
	//SendData();

	//event_base_loopbreak(m_base);

	//LOG4CPLUS_DEBUG(g_logger, "on timeout");
	m_timecost = Now_us() - m_starttick;

	SendFin();
}

void UDPClient::OnTimer(NTDTimer* pTimer)
{
	//write according to the token bucket's limitation
	int total_bytes = 0;
	int total_cnt = 0;
	while(true){
		int32 can_io_bytes = m_send_bucket.GetLimit();
		if( can_io_bytes < m_ptk_size ){
			//LOG4CPLUS_DEBUG(g_logger," tokens not enough: "<<can_io_bytes<<" total sent: "<<m_per_total);
			break;
		} 
		int written = OnWrite();
		if( written > 0 ){
			int left = m_send_bucket.RemoveTokens(written);
			total_bytes += written;
			++total_cnt;
		}else{
			//LOG4CPLUS_DEBUG(g_logger,"written: "<<written<<" tokens: "<<m_send_bucket.GetLimit());
			break;
		}
	}

	if( total_bytes > 0 ){
		LOG4CPLUS_DEBUG(g_logger,"total sent bytes: "<< total_bytes<<" total_cnt: "<<total_cnt);
	}
}

void UDPClient::OnTokenFilled(TokenBucket * pBucket)
{
	LOG4CPLUS_DEBUG(g_logger,"token filled");
	event_add(m_writeEv, NULL);
	//m_tick.Start(1, true);
	OnTimer(NULL);
}

void UDPClient::OnTokenEmpty(TokenBucket * pBucket)
{
	LOG4CPLUS_DEBUG(g_logger,"token empty");
	event_del(m_writeEv);
	//m_tick.Stop();
}


int32 UDPClient::OnWrite()
{
	int ret = 0;
	int written = 0;

	if( m_bFin ){
		if( m_hsInterval > SYN_TIMEOUT){
			//event_base_loopbreak(m_base);
			if( m_pListen ){
				m_pListen->HandleError(IUdpClientListener::ERR_FIN_TIMEOUT);
				return -1;
			}
		}
		else if(Now_us() > m_lastsent + m_hsInterval){
			ret = SendFin();
		}
	}
	else if(!m_bInited){
		if( m_hsInterval > SYN_TIMEOUT){
			//event_base_loopbreak(m_base);
			if( m_pListen ){
				m_pListen->HandleError(IUdpClientListener::ERR_HS_TIMEOUT);
				return -1;
			}
		}
		else if(Now_us() > m_lastsent + m_hsInterval){
			ret = SendHandShake();
		}
	}else{
		ret = SendData();
		LOG4CPLUS_TRACE(g_logger,"sent bytes: "<< ret);
		written += ret>0?ret:0;
		if( ret>0 && m_bPacketPair && m_seq%2 == 0){
			ret = SendData();
			LOG4CPLUS_TRACE(g_logger,"pair, sent bytes: "<< ret);
			written += ret>0?ret:0;
		}

		if( Now_us() - m_starttick >= m_timeout_us ){
			OnFinish();
		}
	}
//	}else if( Now_us() > m_lastsent + m_pktInterval ){
//		//m_totaltick += Now_us() - m_lastsent;
//		SendData();
//		if( m_bPacketPair && m_seq%2 == 0){
//			SendData();
//		}
//
//#ifdef SELF_CLOCK
//		if( Now_us() - m_starttick >= m_timeout_us ){
//			OnTimer();
//		}
//#endif
//	}
//	else{
//		//Sleep(10);
//	}
	
	if( ret < 0 ){
		int err = evutil_socket_geterror(m_socket);
		if(EVUTIL_ERR_RW_RETRIABLE(err) && !event_pending(m_writeEv, EV_WRITE,NULL)){
			LOG4CPLUS_TRACE(g_logger,"add write_event notification");
			event_add(m_writeEv, NULL);
		}		
	}else{
		event_del(m_writeEv);
	}

	return written;
}

void UDPClient::SetRateLimit(uint32 limit)
{
	m_ratelimit = limit;

    m_ptk_size = static_cast<int32>(g_settings.m_probeTime * limit / g_settings.m_samples);
	if(g_settings.m_len != 0){
		m_ptk_size = g_settings.m_len;
	}

	if( m_ptk_size > g_settings.m_mss ){
		m_ptk_size = g_settings.m_mss;
	}

	int32 payload = m_ptk_size - IP_HEAD_LEN - UDP_HEAD_LEN - (m_protodata.GetPaketSize() - m_protodata.m_len);
	if( payload < 0 ){
		payload = 1;
	}else if( payload > g_settings.m_mss ){
		payload = g_settings.m_mss;
	}

	if( m_protodata.m_pData ){
		delete []m_protodata.m_pData;
	}

	m_protodata.m_len = payload;
	m_ptk_size = IP_HEAD_LEN + UDP_HEAD_LEN + m_protodata.GetPaketSize();

	if( payload > 0 ){
		m_protodata.m_pData = new byte[payload];
		memset(m_protodata.m_pData, 0, m_protodata.m_len*sizeof(byte));
	}

	m_pktInterval = 1000000 * m_ptk_size / m_ratelimit;

	if( limit == UINT_MAX ){
		m_send_bucket.Init(NULL);
	}else{
		//m_cfg.rate_limit_ = limit;
		//m_cfg.rate_burst_ = limit;
		m_cfg.Init(limit, 2*limit, 10);
		m_send_bucket.Init(&m_cfg, false, m_ptk_size);
	}
}

int32 UDPClient::SendHandShake()
{
	NtdProtoHandShake handshake;
	handshake.m_ini_seq = NtdSeq::decseq(m_seq);
	handshake.m_convId = m_protodata.m_convId;
	handshake.m_bPacketPair = m_bPacketPair;
	handshake.Encode(m_evbuf_output);

	int8 packet[TEMP_BUFFER];

	//send packets
	int32 length = evbuffer_get_length(m_evbuf_output);
	evbuffer_remove(m_evbuf_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_dest, sizeof(m_dest));

	if( nSendBytes > 0 ){
		m_lastsent = Now_us();
		m_hsInterval = m_hsInterval + m_hsInterval/2;
		m_seq = NtdSeq::decseq(m_seq);
	}

	LOG4CPLUS_DEBUG(g_logger,handshake.Desc());

	return nSendBytes;
}

int32 UDPClient::SendFin()
{
	m_bFin = true;

	NtdProtoFin fin;
	fin.m_convId = m_protodata.m_convId;
	fin.Encode(m_evbuf_output);

	int8 packet[TEMP_BUFFER];

	//send packets
	int32 length = evbuffer_get_length(m_evbuf_output);
	evbuffer_remove(m_evbuf_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_dest, sizeof(m_dest));

	if( nSendBytes > 0 ){
		m_lastsent = Now_us();
		m_hsInterval = m_hsInterval + m_hsInterval/2;
	}

	LOG4CPLUS_DEBUG(g_logger,fin.Desc());

	return nSendBytes;
}

int32 UDPClient::SendData()
{
	//uint64 now = Now_us();
	//m_totaltick += now - m_lastsent;

	m_protodata.m_ptk_seq = NtdSeq::incseq(m_seq);
	m_protodata.m_timestamp_sending = Now_ms();
	m_protodata.Encode(m_evbuf_output);

	int8 packet[TEMP_BUFFER];

	//send packets
	int32 length = evbuffer_get_length(m_evbuf_output);
	evbuffer_remove(m_evbuf_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_dest, sizeof(m_dest));

	if( nSendBytes > 0 ){
		assert(nSendBytes == length);
		m_lastsent = Now_us();
		m_total += m_protodata.GetPaketSize();
		m_per_total += m_protodata.GetPaketSize();
		m_seq = NtdSeq::incseq(m_seq);
	}


	//LOG4CPLUS_DEBUG(g_logger,"now = "<<m_protodata.m_timestamp_sending);

	LOG4CPLUS_TRACE(g_logger,m_protodata.Desc());

	//may casue some re-entry problems
	//if( now - m_starttick >= m_timeout_us ){
	//	OnTimer();
	//}

	return nSendBytes;
}

void UDPClient::OnRead()
{
	int8 packet[TEMP_BUFFER];
	struct sockaddr_in addr;
	ev_socklen_t addrlen;
	int r;

	while(true) {
		addrlen = sizeof(struct sockaddr_in);
		r = recvfrom(m_socket, packet, sizeof(packet),	0, (struct sockaddr*) &addr, &addrlen);
		if (r < 0) {
			int err = evutil_socket_geterror(m_socket);
			if( !UTDUTIL_ERR_RW_RETRIABLE(err) ){
				LOG4CPLUS_ERROR(g_logger,
					"Error while reading, "<<evutil_socket_error_to_string(err));
			}
			return;
		}

		if( r == 0 ){
			LOG4CPLUS_ERROR(g_logger,"Incorrect UDP packet");
			return;
		}

		evbuffer_add(m_evbuf_input, packet, r);

		bool ret = NtdProtoParser::ParseProtocol(m_evbuf_input, NULL);
		if( !ret || evbuffer_get_length(m_evbuf_input) != 0 ){
			LOG4CPLUS_ERROR(g_logger, "Error length while reading, ");
			evbuffer_drain(m_evbuf_input, evbuffer_get_length(m_evbuf_input));
			return;
		}
	}
}

void UDPClient::HandleFinResp(NtdProtoFinResp * proto, void* pContext)
{
	if(m_pListen && m_bFin && m_bInited){
		LOG4CPLUS_DEBUG(g_logger,proto->Desc()<<"\n");
		m_pListen->HandleResult(m_ratelimit, proto->m_recv_rate, proto->m_bCongestion, proto->m_pkt_loss);
	}else{
		//LOG4CPLUS_DEBUG(g_logger,proto->Desc()<<", ignore dup");
	}
	//event_base_loopbreak(m_base);
}

void UDPClient::HandleHandShakeResp(NtdProtoHandShakeResp * proto, void* pContext)
{
	LOG4CPLUS_DEBUG(g_logger,proto->Desc());

	if(!m_bInited){
		m_bInited = true;

		m_hsInterval = SYN_INTERVAL;

		SetDuration(g_settings.m_probeTime);
	}
}

void UDPClient::HandleProtoFeedback(NtdProtoFeedback * proto, void* pContext)
{
	//return;
	//LOG4CPLUS_TRACE(g_logger,proto->Desc());
	uint32 now = Now_ms();
	uint32 rtt = now - proto->m_timestamp_sending;

	int32 owd = proto->m_timestamp_recving - proto->m_timestamp_sending;
	if(m_owd_base == 0){
		m_owd_base = owd;
	}

	int32 owd_var = owd - m_owd_base;

	LOG4CPLUS_DEBUG(g_logger, " conv_id = "<< proto->m_convId << " now = "<< now <<" ms, send_ts = "<<proto->m_timestamp_sending
		<< " ms, rtt = " << rtt <<" ms, rel_owd = " << owd_var <<" ms "
		<<" recv_rate = " << proto->m_recving_rate << "Bytes/sec "
		<<" pkt_loss = " << proto->m_pkt_loss << "%");
}